#include "core.h"
#include "os.h"
#include "http_serv.h"
#include "http_cli.h"
#include "cJSON.h"

#include <string.h>
#include <arpa/inet.h>
#include <sys/stat.h>
#include <sys/wait.h>
#include <errno.h>
#include <pthread.h>
#include <unistd.h>

#define K 768 // any value less than 1 will FATAL !!!
#define MEM_KEPP 1024 * 1024 * 1024 // 1GB
#define ADDR_LIST_PATH "/tmp/meth/____addr.list"

//#return: 'Y' for enough, 'N' for lack
static char _sysmemory_enough(_ui limit);

//@param n[in]: num of random-role nodes to start
static Error * _start_n(_ui n) __mustuse;

//@param ip[in]: node's ipaddr
//@param rolebit[in]: BIT_STATE_ROLE_{FULL,MINER,LIGHT}
//#return: nil for success
static Error *(_start) (char *ip, _i rolebit, const char *bootnode_addr) __prm_nonnull __mustuse;

//@param ip[in]: node's ipaddr
//#return: nil for success
static Error *(_stop) (char *ip) __prm_nonnull __mustuse;

//@param ip[in]: node's ipaddr
//@param msdelay[in]: node network delay in unit[ms]
//#return: nil for success, other value for error
static Error *(_set_delay) (char *ip, _i msdelay) __prm_nonnull __mustuse;

//** MASTER FUNCTION **
static Error *engine(void) __mustuse;

//**public interface**
__pub struct Core core = {
    .engine = engine,
};

//@member ip: 0 for 'uninitialized'
//@member state: 0 for 'stop', 1 for 'running', 2 for 'unknown'
//@param delay: node's delay in ms, out == in
struct Node{
    char *ip;
    _ui state;
    _ui delay;
};

//@member random_switch: global random-ops switch, 'ON' or "OFF"
struct Cluster{
    struct Node buckets[K];
    _i random_switch;
};

static pthread_mutex_t lk = PTHREAD_MUTEX_INITIALIZER;
static pthread_cond_t cv = PTHREAD_COND_INITIALIZER;
struct Cluster cluster;
char *bootnode_addr = "";

#define ON 0
#define OFF 1

#define BIT_STATE_STOP 1
#define BIT_STATE_RUNNING 2

#define BIT_STATE_ROLE_FULL  3
#define BIT_STATE_ROLE_LIGHT 4
#define BIT_STATE_ROLE_MINER 5

static Error *
_start_n(_ui n){
    Error *e;
    _i i, rolebit;
    while(n){
        if('N' == _sysmemory_enough(MEM_KEPP)){
            return __err_new(-1, "system memory lack!", nil);
        }

        switch(utils.urand() % 10){
            case 0:
                rolebit = BIT_STATE_ROLE_MINER;
                break;
            case 1:;
            case 2:;
            case 3:;
                rolebit = BIT_STATE_ROLE_LIGHT;
                break;
            default:;
                rolebit = BIT_STATE_ROLE_FULL;
        }

        i = utils.urand() % K;
        if(! __check_bit(cluster.buckets[i].state, BIT_STATE_RUNNING)){
            pthread_mutex_lock(&lk);
            while(ON != cluster.random_switch){
                pthread_cond_wait(&cv, &lk);
            };
            if(nil != (e = _start(cluster.buckets[i].ip, rolebit, bootnode_addr))){
                pthread_mutex_unlock(&lk);
                return __err_new(-1, "****", e);
            }
            __set_bit(cluster.buckets[i].state, BIT_STATE_RUNNING);
            __set_bit(cluster.buckets[i].state, rolebit);
            pthread_mutex_unlock(&lk);
#ifndef _RELEASE
            __info(cluster.buckets[i].ip);
#endif
        }
        n--;
    }

    return nil;
}

inline static Error *
random_start(void){
    Error *e;
    _ui n = utils.urand() % (10 + utils.urand() % 50);
    __return_if_not_nil(e, _start_n(n));
    return nil;
}

inline static Error *
random_stop(void){
    Error *e;
    _ui n = utils.urand() % (40 + utils.urand() % 200),
        i;
    while(n){
        //never stop the first node with IP 10.0.0.1, keep alive for bootnode
        i = 1 + utils.urand() % (K - 1);

        if(! __check_bit(cluster.buckets[i].state, BIT_STATE_STOP)){
            pthread_mutex_lock(&lk);
            while(ON != cluster.random_switch){
                pthread_cond_wait(&cv, &lk);
            };
            if(nil != (e = _stop(cluster.buckets[i].ip))){
                pthread_mutex_unlock(&lk);
                return __err_new(-1, "****", e);
            }
            cluster.buckets[i].state = 0;
            __set_bit(cluster.buckets[i].state, BIT_STATE_STOP);
            cluster.buckets[i].delay = 0;
            pthread_mutex_unlock(&lk);
#ifndef _RELEASE
            __info(cluster.buckets[i].ip);
#endif
        }
        --n;
    }

    return nil;
}

inline static Error *
random_delay(void){
    Error *e;
    _ui n = utils.urand() % (20 + utils.urand() % 100),
        i,
        msdelay;
    while(n){
        i = utils.urand() % K;
        if(__check_bit(cluster.buckets[i].state, BIT_STATE_RUNNING)){
            pthread_mutex_lock(&lk);
            while(ON != cluster.random_switch){
                pthread_cond_wait(&cv, &lk);
            };
            msdelay = utils.urand() % 1000;
            if(nil != (e = _set_delay(cluster.buckets[i].ip, msdelay))){
                pthread_mutex_unlock(&lk);
                return __err_new(-1, "****", e);
            }
            cluster.buckets[i].delay = msdelay;
            pthread_mutex_unlock(&lk);
#ifndef _RELEASE
            __info(cluster.buckets[i].ip);
#endif
        }
        --n;
    }

    return nil;
}

#ifdef _OS_FREEBSD
static Error *
talk_to_node(char *ip, Source *s){
    __return_err_if_any_nil(ip&&s);
    const char *headers[2] = {"Content-Type", "application/json"};
    char url[sizeof("http://:8545") + strlen(ip)];
    sprintf(url, "http://%s:8545", ip);

    struct HttpReq req;
    req.url = url;
    req.method = "POST";
    req.nheaders_x_2 = 2;
    req.headers = headers;
    req.body.dsiz = s->dsiz;
    req.body.data = s->data;
    req.body.drop = s->drop;

    Error *e;
    __return_if_not_nil(e, httpcli.req(&req));

    s->data = req.body.data;
    s->dsiz = req.body.dsiz;
    s->drop = req.body.drop;

    return nil;
}
#endif

#ifdef _OS_LINUX
static Error *
talk_to_node(char *ip, Source *s){
    __return_err_if_any_nil(ip&&s);

    char ipc_path[sizeof("/tmp/meth/linux/home//eth.ipc") + strlen(ip)];
    sprintf(ipc_path, "/tmp/meth/linux/home/%s/eth.ipc", ip);

    Error *e;
    _i unixsock = 0;
    __set_bit(unixsock, _UNIX_SOCKET_BIT_IDX);
    __return_if_not_nil(e, os.connect(ipc_path, nil, &unixsock));

    if(nil != (e = os.send(unixsock, s->data, s->dsiz))){
        close(unixsock);
        return __err_new(-1, "****", e);
    }

    s->drop(s);

    s->dsiz = 8 * 1024;
    s->data = __alloc(s->dsiz);
    s->drop = utils.sys_drop;
    if(nil != (e = os.recv(unixsock, s->data, &s->dsiz))){
        close(unixsock);
        return __err_new(-1, "****", e);
    }

    close(unixsock);

    return nil;
}
#endif

//**use parity as bootnode on linux, or use geth**
static Error *
get_bootnode_addr(void){
    Error *e;
#ifdef _OS_FREEBSD
    Source s = {
        .data = "{\"jsonrpc\":\"2.0\",\"method\":\"admin_nodeInfo\",\"params\":[],\"id\":4}",
        .dsiz = sizeof("{\"jsonrpc\":\"2.0\",\"method\":\"admin_nodeInfo\",\"params\":[],\"id\":4}") - 1,
        .drop = utils.non_drop,
    };
#endif

#ifdef _OS_LINUX
    Source s = {
        .data = "{\"method\":\"parity_enode\",\"params\":[],\"id\":4,\"jsonrpc\":\"2.0\"}",
        .dsiz = sizeof("{\"method\":\"parity_enode\",\"params\":[],\"id\":4,\"jsonrpc\":\"2.0\"}") - 1,
        .drop = utils.non_drop,
    };
#endif

    __return_if_not_nil(e, talk_to_node(cluster.buckets[0].ip, &s));

    cJSON *root = cJSON_Parse(s.data);

#ifdef _OS_FREEBSD
    cJSON *result = cJSON_GetObjectItemCaseSensitive(root, "result");
    cJSON *enode = cJSON_GetObjectItemCaseSensitive(result, "enode");
#endif

#ifdef _OS_LINUX
    cJSON *enode = cJSON_GetObjectItemCaseSensitive(root, "result");
#endif

    if(cJSON_IsString(enode) && nil != enode->valuestring){
        bootnode_addr = __alloc(strlen(enode->valuestring));
        strcpy(bootnode_addr, enode->valuestring);

        s.drop(&s);
        cJSON_Delete(root);
    } else {
        s.drop(&s);
        cJSON_Delete(root);

        return __err_new(-1, "cJSON parse error", nil);
    }

    return nil;
}

static Error *
preinit(void){
    cluster.random_switch = ON;
    for(_i i = 0; i < K; ++i){
        cluster.buckets[i].ip = nil;
        __set_bit(cluster.buckets[i].state, BIT_STATE_STOP);
    }

    Error *e;
    Source s;//need NOT free
    char *orig;
    _i cnt;
    _i lastindex;

    __return_if_not_nil(e, os.read_file(ADDR_LIST_PATH, &s));

    orig = s.data;
    cnt = 0;
    lastindex = 0;
    for(_ui i = 0; cnt < K && i < s.dsiz; ++i){
        if(' ' == orig[i]){
            orig[i] = '\0';
            cluster.buckets[cnt].ip = orig + lastindex;
            __set_bit(cluster.buckets[cnt].state, BIT_STATE_STOP);
            lastindex = i + 1;
            ++cnt;
        }
    }

    if(K > cnt){
        e = __err_new(-1, "num of nodes NOT match (need more)!", nil);
    }

    // start the persistent bootnode
    __return_if_not_nil(e, _start(cluster.buckets[0].ip, BIT_STATE_ROLE_FULL, ""));

    //wait bootnode to produce its' enode addr
    sleep(10);
    __return_if_not_nil(e, get_bootnode_addr());

    // random start [10, 50) nodes
    __return_if_not_nil(e, _start_n(10 + utils.urand() % 40));
    return nil;
}

//** random running engine **
static Error *
engine(void){
    Error *e;
    Error *(*ops[3]) (void) = {
        random_stop,
        random_delay,
        random_start,
    };

    __return_if_not_nil(e, preinit());

    // random-select one ops per [1, 10]s
    for(_ui i;;){
        utils.msleep(1000 + utils.urand() % 10000);
        i = utils.urand() % 3;

        if(nil != (e = ops[i]())){
            __display_and_clean(e);
        }
    }

    return nil;
}

/////////////////////////////////////////////////////////////
/////////////////////////////////////////////////////////////

//@param ip[in]: node's ipaddr
//#return: nil for success, other value for error
//TODO: role impl in helper-scripts
static Error *
_start (char *ip, _i role __unuse, const char *bootnode_addr){
    pid_t pid = fork();
    if(0 == pid){
        //fill arg0 with 'ip', cat be used to kill nodes correctly on cmdline
        if(-1 == execl("/tmp/meth/helper/_start", ip, ip, bootnode_addr, nil)){
            __fatal_sys();
        }
    } else if(0 > pid){
        return __err_new(errno, strerror(errno), nil);
    }

    waitpid(-1, nil, WNOHANG);
    return nil;
}

//@param ip[in]: node's ipaddr
//#return: nil for success, other value for error
static Error *
_stop (char *ip){
    pid_t pid = fork();
    if(0 == pid){
        //fill arg0 with 'ip', cat be used to kill nodes correctly on cmdline
        if(-1 == execl("/tmp/meth/helper/_stop", ip, ip, nil)){
            __fatal_sys();
        }
    } else if(0 > pid){
        return __err_new(errno, strerror(errno), nil);
    }

    waitpid(-1, nil, WNOHANG);
    return nil;
}

//@param ip[in]: node's ipaddr
//@param delay_ms[in]: node network delay in millisecond
//#return: nil for success, other value for error
static Error *
_set_delay (char *ip, _i msdelay){
    pid_t pid = fork();
    if(0 == pid){
        char msdelaystr[4];
        sprintf(msdelaystr, "%d", msdelay);
        msdelaystr[3] = '\0';
        //fill arg0 with 'ip', cat be used to kill nodes correctly on cmdline
        if(-1 == execl("/tmp/meth/helper/_set_delay", ip, ip, msdelaystr, nil)){
            __fatal_sys();
        }
    } else if(0 > pid){
        return __err_new(errno, strerror(errno), nil);
    }

    waitpid(-1, nil, WNOHANG);
    return nil;
}

#ifdef _OS_FREEBSD
//#return: 'Y' for enough, 'N' for lack
static char
_sysmemory_enough(_ui limit){
    //TODO
    _ui avaliable_mem = 1 + limit;
    if(limit > avaliable_mem){
        return 'N';
    }
    return 'Y';
}
#endif

#ifndef _OS_FREEBSD
//#return: 'Y' for enough, 'N' for lack
static char
_sysmemory_enough(_ui limit){
    //TODO
    _ui avaliable_mem = 1 + limit;
    if(limit > avaliable_mem){
        return 'N';
    }
    return 'Y';
}
#endif

/***************
 * HTTP SERVER *
 ***************/
//**hdr: random-ops stop**
//http://[ADDR]/cluster/random/enable
//@param reqbody[in]: user need NOT care
//@param reqbody_siz[in]: user need NOT care
//@param resp[out]: user need NOT care
//#return http_status_code
static _i
random_disable_hdr(void *reqbody __unuse, size_t reqbody_siz __unuse, Source *resp){
    pthread_mutex_lock(&lk);
    cluster.random_switch = OFF;
    pthread_mutex_unlock(&lk);
    //**NEED NOT** pthread_cond_signal(&cv);

    resp->data = "{\"result\":\"SUCCESS\"}";
    resp->dsiz = sizeof("{\"result\":\"SUCCESS\"}") - 1;
    resp->drop = utils.non_drop;

    return 200;
}

//**hdr: random-ops start**
//http://[ADDR]/cluster/random/disable
//@param reqbody[in]: user need NOT care
//@param reqbody_siz[in]: user need NOT care
//@param resp[out]: user need NOT care
//#return http_status_code
static _i
random_enable_hdr(void *reqbody __unuse, size_t reqbody_siz __unuse, Source *resp){
    pthread_mutex_lock(&lk);
    cluster.random_switch = ON;
    pthread_mutex_unlock(&lk);
    pthread_cond_signal(&cv);

    resp->data = "{\"result\":\"SUCCESS\"}";
    resp->dsiz = sizeof("{\"result\":\"SUCCESS\"}") - 1;
    resp->drop = utils.non_drop;

    return 200;
}

//**hdr: running-nodes query**
//http://[ADDR]/cluster/nodes/query
//@param reqbody[in]: user need NOT care
//@param reqbody_siz[in]: user need NOT care
//@param resp[out]: user need NOT care
//#return http_status_code
static _i
nodes_query_hdr(void *reqbody __unuse, size_t reqbody_siz __unuse, Source *resp){
    const char *role;
    //json form LIKE: [{"id":"999","role":"miner","delay":"120ms"]}]
    //'id' is the node's index in the cluster.buckets
    size_t bufsiz = sizeof("{\"id\":\"999\",\"role\":\"miner\",\"delay\":\"120ms\"]},") * K;

    resp->data = __alloc(bufsiz);
    resp->drop = utils.sys_drop;
    resp->dsiz = sprintf(resp->data, "[");
    for(_i i = 0; i < K; ++i){
        if(__check_bit(cluster.buckets[i].state, BIT_STATE_RUNNING)){
            if(__check_bit(cluster.buckets[i].state, BIT_STATE_ROLE_LIGHT)){
                role = "light";
            } else if (__check_bit(cluster.buckets[i].state, BIT_STATE_ROLE_MINER)){
                role = "miner";
            } else {
                role = "full";
            }
            resp->dsiz +=
                snprintf(resp->data + resp->dsiz, bufsiz - resp->dsiz,
                        "{\"id\":\"%d\",\"role\":\"%s\",\"delay\":\"%dms\"},",
                        i, role, cluster.buckets[i].delay);
        }
    }

    if((sizeof("[") - 1) == resp->dsiz){
        ((char *)resp->data)[sizeof("[") - 1] = ']';
        resp->dsiz += sizeof("]") - 1;
    } else {
        ((char *)resp->data)[resp->dsiz - 1] = ']';
    }

    return 200;
}

//**hdr: forcing all nodes to be customized**
//http://[ADDR]/cluster/nodes/custom
//@param reqbody[in]: user need NOT care
//@param reqbody_siz[in]: user need NOT care
//@param resp[out]: user need NOT care
//#return http_status_code
static _i
nodes_custom_hdr(void *reqbody, size_t reqbody_siz, Source *resp){
    if(!reqbody || 0 == reqbody_siz){
        return 400;
    }

    resp->dsiz = sizeof("this API is in 'TODO'ing") - 1;
    resp->data = "this API is in 'TODO'ing";
    resp->drop = utils.non_drop;

    //TODO
    //users provide a global node configuration,
    //and this interface will do its best to complete.
    //If it fails, the real-time status will be returned without rollback.
    //
    //==> first, stop global random-ops
    //==> second, calculates the number of nodes currently in each state
    //==> final, do _start OR _stop OR _set_delay to fit the request,
    //    if any ops failed, return the realtime configuration

    return 200;
}

//**hdr: USE TO SUPPORT meth-web**
//http://[ADDR]/cluster/nodes/custom
//@param reqbody[in]: user need NOT care
//@param reqbody_siz[in]: user need NOT care
//@param resp[out]: user need NOT care
//#return http_status_code
static _i
nodes_directrpc_hdr(void *reqbody, size_t reqbody_siz, Source *resp){
    if(!reqbody || 0 == reqbody_siz){
        return 400;
    }

    resp->dsiz = sizeof("this API is in 'TODO'ing") - 1;
    resp->data = "this API is in 'TODO'ing";
    resp->drop = utils.non_drop;

    //TODO
    //use this API to communicate with any nodes
    //batch + different requests MUST be supported !
    //

    return 200;
}

__gen_http_hdr(random_enable, random_enable_hdr);
__gen_http_hdr(random_disable, random_disable_hdr);
__gen_http_hdr(nodes_query, nodes_query_hdr);
__gen_http_hdr(nodes_custom, nodes_custom_hdr);
__gen_http_hdr(nodes_directrpc, nodes_directrpc_hdr);

__init static void
hdr_register(void){
    Error *e;
    __fatal_if_not_nil(e, httpserv.hdr_register("/cluster/random/disable", random_disable, "POST"));
    __fatal_if_not_nil(e, httpserv.hdr_register("/cluster/random/enable", random_enable, "POST"));
    __fatal_if_not_nil(e, httpserv.hdr_register("/cluster/nodes/query", nodes_query, "POST"));
    __fatal_if_not_nil(e, httpserv.hdr_register("/cluster/nodes/custom", nodes_custom, "POST"));
    __fatal_if_not_nil(e, httpserv.hdr_register("/cluster/nodes/directrpc", nodes_directrpc, "POST"));
}

#undef BIT_STATE_RUNNING
#undef BIT_STATE_STOP
#undef MEM_KEPP
#undef K
#undef ADDR_LIST_PATH
